package my.thesis.analytics;

import my.thesis.analytics.metrics.MyMetricsConsumer;
import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.StormSubmitter;
import backtype.storm.generated.AlreadyAliveException;
import backtype.storm.generated.InvalidTopologyException;
import backtype.storm.generated.StormTopology;
import backtype.storm.metric.LoggingMetricsConsumer;

public class App 
{
	//private static final int maxSpoutPending = 100000;
	//private static final int messageTimeout = 100;
	
	private static final int timeIntervalSec = 7200; 	
	
	private static final int timeToStoreHashtags = 10; 
	private static final int timeToStoreCoOccurences = 12 ; 
	
	private static final int timeToStoreSentiment = 10; 
	private static final int timeToStoreMention = 12; 
	
	private static final int numWorkers = 2;
		
	/********* BASELINE 1 **************************/
	
	private static final int fieldExtractorBoltParallelism=1;
	private static final int fieldExtractorBoltNumTasks=1;
	
	private static final int timeIntervalBoltParallelism=1;
	private static final int timeIntervalBoltNumTasks=1;
	
	private static final int counterParallelism=1;
	private static final int counterNumTasks=1;
	
    private static final int counterParallelism2=1;
	private static final int counterNumTasks2=1;
	
	/********* BASELINE 1 **************************/
		
	/********* OPTIMIZED 1 **************************/
	
/*    private static final int fieldExtractorBoltParallelism=4;
	private static final int fieldExtractorBoltNumTasks=4;
	
	private static final int timeIntervalBoltParallelism=4;
	private static final int timeIntervalBoltNumTasks=4;
	
	private static final int counterParallelism=4;
	private static final int counterNumTasks=4;
	
    private static final int counterParallelism2=2;
	private static final int counterNumTasks2=2;*/
	
	/********* OPTIMIZED 1 **************************/

	/********* BASELINE 2 **************************/

/*	private static final int fieldExtractorBoltParallelism=1;
	private static final int fieldExtractorBoltNumTasks=1;
	
	private static final int filterParallelism=1;
	private static final int filterNumTasks=1;
	
	private static final int sentimentParallelism=1;
	private static final int sentimentNumTasks=1;
	
	private static final int wordExtractParallelism=1;
	private static final int wordExtractNumTasks=1;
	
	private static final int counterParallelism=1;
	private static final int counterNumTasks=1;
	
    private static final int counterParallelism2=1;
	private static final int counterNumTasks2=1;
	
	private static final int countryDetectorBoltParallelism=1;
	private static final int countryDetectorBoltNumTasks=1;*/
	
	/********* BASELINE 2 **************************/
	
	/********* OPTIMIZED 2 **************************/
	
/*	private static final int fieldExtractorBoltParallelism=4;
	private static final int fieldExtractorBoltNumTasks=4;
	
	private static final int wordExtractParallelism=4;
	private static final int wordExtractNumTasks=4;
	
	private static final int sentimentParallelism=2;
	private static final int sentimentNumTasks=2;
	
	private static final int filterParallelism=1;
	private static final int filterNumTasks=1;
	
	private static final int counterParallelism=4;
	private static final int counterNumTasks=4;
	
	private static final int counterParallelism2=2;
	private static final int counterNumTasks2=2;
	
	private static final int countryDetectorBoltParallelism=1;
	private static final int countryDetectorBoltNumTasks=1;*/
	
	/********* OPTIMIZED 2 **************************/
	
	private static String webserverurl= "http://192.168.1.101:80/MY_TWEETS_HASHTAGS_29_30_08_40.json?1";
	private static String databaseurl="192.168.1.101";
	private static Boolean localmode=false;
/*		
	private static String webserverurl= "http://150.140.16.45:80/MY_TWEETS_HASHTAGS_29_30_08_40.json?1";
	private static String databaseurl="150.140.16.45";
	private static Boolean localmode=false;
	private static String webserverurl= "http://localhost:80/MY_TWEETS_HASHTAGS_29_30_08_40.json?1";
	private static String databaseurl="localhost";
	private static Boolean localmode=true;
*/
	
    public static void main( String[] args ) throws InterruptedException, AlreadyAliveException, InvalidTopologyException
    {

	//********************************** SINGLE SPOUT TOPOLOGY GIA EGGRAFH SE TEXT *********************************************
	//TopologyBuilder builder = new TopologyBuilder();
	//builder.setSpout("TweetReaderSpout", new TweetReader());
	//StormTopology topology =builder.createTopology();
	//*********************************** SINGLE SPOUT TOPOLOGY GIA EGGRAFH SE TEXT ********************************************
	    	
    StormTopology topology=TopologyBuilderUtil.buildTopo1(timeIntervalSec,timeToStoreHashtags,timeToStoreCoOccurences
   				 ,fieldExtractorBoltParallelism,fieldExtractorBoltNumTasks,timeIntervalBoltParallelism,timeIntervalBoltNumTasks,
   				 counterParallelism, counterNumTasks,counterParallelism2, counterNumTasks2,webserverurl,databaseurl);  
        	
    /*    	
        StormTopology topology= TopologyBuilderUtil.buildTopo2(timeToStoreSentiment,timeToStoreMention
   			 ,fieldExtractorBoltParallelism, fieldExtractorBoltNumTasks, filterParallelism,filterNumTasks,
   			 	countryDetectorBoltParallelism,countryDetectorBoltNumTasks, counterParallelism,counterNumTasks,
   			 	counterParallelism2, counterNumTasks2,wordExtractParallelism,wordExtractNumTasks,sentimentParallelism,sentimentNumTasks,
   			 	webserverurl,databaseurl);*/
   
    /* 
     	StormTopology topology=TopologyBuilderUtil.buildTopo1Reliable(timeIntervalSec,timeToStoreHashtags,timeToStoreCoOccurences
			 ,fieldExtractorBoltParallelism,fieldExtractorBoltNumTasks,timeIntervalBoltParallelism,timeIntervalBoltNumTasks,
			 counterParallelism, counterNumTasks,counterParallelism2, counterNumTasks2,webserverurl,databaseurl);
  */
    /*    	
     StormTopology topology= TopologyBuilderUtil.buildTopo2Reliable(timeToStoreSentiment,timeToStoreMention
			 ,fieldExtractorBoltParallelism, fieldExtractorBoltNumTasks, filterParallelism,filterNumTasks,
			 	countryDetectorBoltParallelism,countryDetectorBoltNumTasks, counterParallelism,counterNumTasks,counterParallelism2,counterNumTasks2,
			 	webserverurl,databaseurl);*/
  
       
    Config conf = new Config();
	conf.setDebug(false);
	
	conf.registerMetricsConsumer(MyMetricsConsumer.class, 1);
					
	//------------------ ΕΔΩ ΝΑ ΒΑΛΩ CONFIGURATIONS-------------------------
	
    conf.setNumWorkers(numWorkers);
	//conf.setMaxSpoutPending(maxSpoutPending);
	//conf.setMessageTimeoutSecs(messageTimeout);

	//------------------ ΕΔΩ ΝΑ ΒΑΛΩ CONFIGURATIONS-------------------------
		
		if(localmode)
		{
		
			LocalCluster cluster = new LocalCluster();
			cluster.submitTopology("mytopology", conf, topology);
		}
		else
		{
			StormSubmitter.submitTopology("mytopology", conf, topology);
		}
    }
}
